package cn.lagou.text

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

object ClickImp {

  def main(args: Array[String]): Unit = {
    //创建sparkconntext
    val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName(this.getClass.getCanonicalName)
      .getOrCreate()
    val sc = spark.sparkContext

    val clickRDD: RDD[String] = sc.textFile("data\\clik.log")
    val impRDD: RDD[String] = sc.textFile("data\\imp.log")

    val clickRes = clickRDD.map{line => {
      val arr = line.split("\\s+")
      val adid = arr(3).substring(arr(3).lastIndexOf("=") + 1)
      (adid, 1)
    }}.reduceByKey(_ + _)

    val impRes = impRDD.map { line =>
      val arr = line.split("\\s+")
      val adid = arr(3).substring(arr(3).lastIndexOf("=") + 1)
      (adid, 1)
    }.reduceByKey(_ + _)

    //保存到hdfs
    val res: RDD[String] = clickRes.fullOuterJoin(impRes)
      .map(x => x._1 + "," + x._2._1.getOrElse(0) + "," + x._2._2.getOrElse(0))
      .repartition(1)

    res
      //.foreach(println)
      //.saveAsTextFile("hdfs://linux145:9000/data/")
      .saveAsTextFile("data/add_log")

    spark.close()
    //sc.stop()
  }

}
